package com.bkd.report

import com.bkd.beans.Log
import com.bkd.tools.ReqUtil
import com.bkd.util.RptUtils
import org.apache.commons.lang.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object AppRpt {

  def main(args: Array[String]): Unit = {
    if(args.length != 3){
      println(
        """
          |com.bkd.report.AppRpt
          |参数
          |数据字典路径
          |输入文件路径
          |输出文件路径
        """.stripMargin)
      sys.exit()
    }

    //制定接受的参数
    var Array(dictFilePath,inputPath,outputPath)=args
    //创建sparkconf
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local[*]")
    //制定序列化方式
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //创建sparkContext
    val sc = new SparkContext(conf)


    //读取数据字典文件
    val dictMap: Map[String,String] = sc.textFile(dictFilePath).map(line => {
      val fields: Array[String] = line.split(":")
      (fields(0),fields(1))
    }).collect().toMap
    //广播数字字典文件
    val broadcast: Broadcast[Map[String, String]] = sc.broadcast(dictMap)
    //读取日志文件
    sc.textFile(inputPath)
      .map(_.split(",",-1))
      .filter(_.length>=85)
    //判断appid  appname不为空
      .map(Log(_))
      .filter(log => !log.appid.isEmpty || !log.appname.isEmpty)
      .map(log => {
        //获取appname  如果name是空的 就去广播文件中 通过appid 获取到appname
        val appName: String = log.appname
        if(StringUtils.isNotEmpty(appName)){
          //从广播中获取到名称
          broadcast.value.getOrElse(log.appid,"未知")
        }
        //计算其他指标
        val reqlist: List[Double] = ReqUtil.caculateReq(log.requestmode,log.processnode)
        val parceList: List[Double] = ReqUtil.caculatePrc(log.iseffective,log.isbilling,log.isbid,log.adorderid,log.iswin,log.winprice,log.adpayment)
        val clickList: List[Double] = ReqUtil.caculateClick(log.requestmode,log.iseffective)
        //组合元组
        (appName,reqlist++parceList++clickList)

      }).reduceByKey((list1,list2)=>{
      list1.zip(list2).map(t=>t._1+t._2)
    }).saveAsTextFile(outputPath)
    sc.stop()
  }
}
